package zju.dgm.service.dataservice;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import zju.dgm.dto.DataFrame;
import zju.dgm.util.Serializer;
import zju.dgm.util.SyncUtil;

import java.util.Deque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Condition;

/**
 * @author lzy
 */
public class NettyServerThread extends Thread {

    private static final Logger logger = LoggerFactory.getLogger(NettyServerThread.class);

    int port;
    Deque<Subscriber> subscribers;


    public EventLoopGroup bossGroup;
    public EventLoopGroup workerGroup;
    ServerBootstrap bootstrap;

//    CyclicBarrier barrier = null;
//    Set<Thread> waiters = ConcurrentHashMap.newKeySet();
//    Deque<DataFrame> dataQueue = new ConcurrentLinkedDeque<>();
//    final Object lock = new Object();


    public NettyServerThread(int port, Deque<Subscriber> subscriber) {
        this.port = port;
        this.subscribers = subscriber;
    }


    @Override
    public void run() {
        //创建两个线程组 boosGroup、workerGroup
        while (true) {
            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
            try {
                //创建服务端的启动对象，设置参数
                bootstrap = new ServerBootstrap();
                //设置两个线程组boosGroup和workerGroup
                bootstrap.group(bossGroup, workerGroup)
                        //设置服务端通道实现类型
                        .channel(NioServerSocketChannel.class)
                        //设置线程队列得到连接个数
                        .option(ChannelOption.SO_BACKLOG, 128)
                        //设置保持活动连接状态
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        //使用匿名内部类的形式初始化通道对象
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                //给pipeline管道设置处理器
                                socketChannel.pipeline().addLast(new MyServerHandler(subscribers));
                            }
                        });//给workerGroup的EventLoop对应的管道设置处理器4
                //绑定端口号，启动服务端
                ChannelFuture channelFuture = bootstrap.bind(port).sync();
                DataService.finLatch.countDown();
                //对关闭通道进行监听
                channelFuture.channel().closeFuture().sync();

            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                logger.info("发生异常，关闭通道,shotdown处");
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

    class MyServerHandler extends ChannelInboundHandlerAdapter {
        Deque<Subscriber> subscribers;


        MyServerHandler(Deque<Subscriber> subscribers) {
            this.subscribers = subscribers;
        }


        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //获取客户端发送过来的消息
            ByteBuf byteBuf = (ByteBuf) msg;
            DataFrame df = null;
            logger.info("得到了消息");
            byte[] arr;
            if (byteBuf.hasArray()) {// 以byte为内存缓冲
                logger.info("有byte缓冲");
                arr = byteBuf.array();
                int offset = byteBuf.arrayOffset() + byteBuf.readerIndex();
                int len = byteBuf.readableBytes();
                df = Serializer.getObjectFromJSON(new String(arr), DataFrame.class);
            } else {// 直接buffer，需要创建数组来接
                logger.info("无byte缓冲");
                int length = byteBuf.readableBytes();
                arr = new byte[length];
                byteBuf.getBytes(byteBuf.readerIndex(), arr);
                df = Serializer.getObjectFromJSON(new String(arr), DataFrame.class);

            }

            assert df != null;
            for (Subscriber subscriber : subscribers) {
                DataFrame finalDf = df;
                SyncUtil.tpe.execute(() -> {
                    if (subscriber.filter.isWanted(finalDf)) {
                        synchronized (subscriber.lock) {
                            subscriber.df = finalDf;
                            subscriber.lock.notifyAll();
                        }
                    }
                });
            }
        }


        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            //发生异常，关闭通道
            logger.info("发生异常，关闭通道，通道异常处");
            ctx.close();
        }
    }


}
